package org.nbict.iot.task.transit;

import com.alibaba.fastjson.JSON;
import org.apache.commons.lang.StringUtils;
import org.apache.ibatis.io.Resources;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.apache.ibatis.session.SqlSessionFactoryBuilder;
import org.apache.storm.trident.operation.BaseFunction;
import org.apache.storm.trident.operation.TridentCollector;
import org.apache.storm.trident.operation.TridentOperationContext;
import org.apache.storm.trident.tuple.TridentTuple;
import org.apache.storm.tuple.Values;
import org.nbict.iot.protocol.config.ProtocolMapper;
import org.nbict.iot.protocol.context.FieldTypeContext;
import org.nbict.iot.protocol.context.ParserContext;
import org.nbict.iot.protocol.type.ProtocolField;
import org.nbict.iot.protocol.util.BytesOperator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.util.Map;

/**
 * Created by songseven on 18/7/6.
 */
public class BinaryParser extends BaseFunction {

    private static final Logger log = LoggerFactory.getLogger(BinaryParser.class);

    private ProtocolField rootField;
//    private final  byte[] tag = {0x01};
//    private final  byte[] patch = {0x23};
    private int count;
    private int total;
    //获取协议配置信息
    @Override
    public void prepare(Map conf, TridentOperationContext context) {
        count = 0;
        total = 0;
        String resource = "mybatis-config.xml";
        SqlSession session = null;
        try {
            InputStream inputStream = Resources.getResourceAsStream(resource);
            SqlSessionFactory sqlSessionFactory = new SqlSessionFactoryBuilder().build(inputStream);
            session = sqlSessionFactory.openSession();
            ProtocolMapper mapper = session.getMapper(ProtocolMapper.class);
            rootField = mapper.getProtocolFieldById(120);
        } catch (IOException e) {
            //log
        } finally {
            session.close();
        }
    }

    @Override
    public void execute(TridentTuple tuple, TridentCollector
            collector) {
        String topic = StringUtils.replaceOnce((String)tuple.getValue(0), "raw", "transit");

        log.debug("topic => " + topic); //debug

        byte[] raw = (byte[]) tuple.getValue(1);
        byte[] checksum = BytesOperator.splitBytes(raw, 0, 1);

        switch (checksum[0]) {
            case (byte)0x01: //校验成功
            case (byte)0x03: //校验成功但是企业平台未登入
            //case (byte)0x02:
                raw = BytesOperator.splitFrontBytes(raw, 1); //去除校验位
            case (byte)0x23: //打补丁
                try {
                    ParserContext context = new FieldTypeContext(raw, rootField);
                    JSON re = (JSON) context.type().process(context);
                    collector.emit(new Values(topic, re.toJSONString
                            (), true));
                } catch (Exception e) {
                    collector.emit(new Values(topic, raw, false)); //error
                }
                break;
            default:
//                System.out.println("error 1 =>" + BytesOperator.bytesToHexString(raw));
                collector.emit(new Values(topic, raw, false)); //error
        }
    }
}
